package com.rem.flink.flink10Sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author Rem
 * @date 2022-11-09
 */
public class KafkaTableTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        /**
         * 输入输入
         */
        String tableName = "eventTable";
        String createDdl = "CREATE TABLE " + tableName + " (" +
                " `user` STRING, " +
                " url STRING, " +
                " ts TIMESTAMP(3) METADATA FROM 'timestamp' " +
                ") WITH (" +
                " 'connector' = 'kafka', " +
                " 'topic' = 'clicks', " +
                " 'properties.bootstrap.servers' = '10.224.18.157:9092,10.224.18.158:9092,10.224.18.159:9092', " +
                " 'properties.group.id' = 'consumer-group', " +
                " 'scan.startup.mode' = 'latest-offset', " +
                " 'format' =  'json' " +
                ")";
        tableEnv.executeSql(createDdl);

        Table table = tableEnv.sqlQuery("select * from " + tableName);
        tableEnv.toChangelogStream(table).print();


        /**
         * 数据输出
         */
        Table aggResult = tableEnv.sqlQuery("select `user`, count(url) as cnt  from " + tableName + " group by `user`");
        tableEnv.toChangelogStream(aggResult).print();

        String aggResultTable = " CREATE TABLE aggResultTable (" +
                " `user` STRING," +
                " cnt BIGINT," +
                " PRIMARY KEY (`user`) NOT ENFORCED" +
                ") WITH (" +
                " 'connector' = 'upsert-kafka'," +
                " 'topic' = 'clicks-cnt'," +
                " 'properties.bootstrap.servers' = '10.224.18.157:9092,10.224.18.158:9092,10.224.18.159:9092', " +
                " 'key.format' = 'json'," +
                " 'value.format' = 'json'" +
                ")   ";
        tableEnv.executeSql(aggResultTable);

        aggResult.executeInsert("aggResultTable");

        env.execute();
    }
}
